package com.example.flow

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.cancel
import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.cancellable
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.transform
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import kotlinx.coroutines.withTimeoutOrNull
import org.junit.Test
import kotlin.system.measureTimeMillis

class CoroutineTest01 {

    private fun simpleList(): List<Int> = listOf(1, 2, 3)

    //序列 同步的返回多个值
    private fun simpleSequence(): Sequence<Int> = sequence {
        for (i in 1..3) {
            Thread.sleep(1000)
//            yield(i)
            yield(10)
        }
    }

    //是异步的，一次性返回了多个值
    suspend fun simpleList2(): List<Int> {
        delay(1000)
        return listOf<Int>(1, 2, 3)
    }

    suspend fun simpleFLow() = flow<Int> {
        for (i in 1..3) {
            delay(1000)//模式异步操作
            emit(i)
        }
    }

    @Test
    fun `test multiple values`() {
//        simpleList().forEach { value -> println(value) }
        simpleSequence().forEach { value -> println(value) }
    }

    @Test
    fun `test multiple values2`() = runBlocking<Unit> {
        simpleList2().forEach { value -> println(value) }
    }

    @Test
    fun `test multiple values3`() = runBlocking<Unit> {
        launch {
            for (i in 1..3) {
                //这里可以继续做事情
                delay(1000)
                println("搞事情---$i")
            }
        }
        simpleFLow().collect { value -> println(value) }
    }


    fun simpleFLow2() = flow<Int> {
        println("Flow started")
        for (i in 1..3) {
            delay(1000)//模式异步操作
            emit(i)
        }
    }

    /**
     * 冷流
     * 流是按顺序执行的，除非操作了特殊的函数改变了序列
     */
    @Test
    fun `test flow is cold`() = runBlocking<Unit> {
        val flow = simpleFLow2()
        println("Calling collect...")
        flow.collect { value -> println(value) }
        println("Calling collect again...")
        flow.collect { value -> println(value) }
    }

    @Test
    fun `test flow continueAction`() = runBlocking<Unit> {
        (1..5).asFlow().filter {
            it % 2 == 0
        }.map {
            "String $it"
        }.collect {
            println("Collect $it")
        }
    }

    @Test
    fun `test flow builder`() = runBlocking<Unit> {
        flowOf("one", "tow", "three")
            .onEach { delay(1000) }
            .collect { println("Collect $it") }
        (1..5).asFlow().collect { println("Collect $it") }
    }

    fun simpleFLow3() = flow<Int> {
        println("Flow started ${Thread.currentThread().name}")
        for (i in 1..3) {
            delay(1000)//模式异步操作
            emit(i)
        }
    }

    @Test
    fun `test flow context`() = runBlocking<Unit> {
        simpleFLow4().collect { println("Collect $it ${Thread.currentThread().name}") }
//        simpleFLow3().collect { println("Collect $it ${Thread.currentThread().name}") }
    }

    /**
     * 这样会报错
     */
    fun simpleFLow4() = flow<Int> {
        withContext(Dispatchers.IO) {
            for (i in 1..3) {
                delay(1000)//模式异步操作
                emit(i)
            }
        }
    }

    fun simpleFLow5() = flow<Int> {
        println("Flow started ${Thread.currentThread().name}")
        for (i in 1..3) {
            delay(1000)//模式异步操作
            emit(i)
        }
    }.flowOn(Dispatchers.Default)//流的上下文切换

    /**
     * 流的上下文切换调用
     */
    @Test
    fun `test flow on`() = runBlocking<Unit> {
        simpleFLow5().collect { println("Collect $it ${Thread.currentThread().name}") }
    }

    /****事件***/
    fun events() = (1..3)
        .asFlow()
        .onEach { delay(100) }
        .flowOn(Dispatchers.Default)

    @Test
    fun `test flow launch`() = runBlocking<Unit> {
        val job = events()
            .onEach { event -> println("event=$event ${Thread.currentThread().name}") }
//            .collect{} //假如收集的时候想要去其他协程执行用launchIn
//            .launchIn(CoroutineScope(Dispatchers.IO))event=1 //DefaultDispatcher-worker-1 @coroutine#2
            .launchIn(this)//传this 就可以成为runBlocking的协程 event=1 Test worker @coroutine#2
//            .join()//调用join是因为该切换的协程并非runBlocking的子协程，
        // 所以不会等待它执行完，要想执行完，就要join让runBlocking等待
        delay(200)
        job.cancelAndJoin()
    }


    fun simpleFLow6() = flow<Int> {
        for (i in 1..3) {
            delay(1000)//模式异步操作
            emit(i)
            println("Emitting $i")
        }
    }

    /**
     * 流的超时取消
     */
    @Test
    fun `test cancel flow`() = runBlocking<Unit> {
        val job = withTimeoutOrNull(2500) {
            simpleFLow6().collect { value -> println("value=$value") }
        }
        println("完成")
    }


    fun simpleFLow7() = flow<Int> {
        for (i in 1..5) {
            emit(i)//ensureActive 会被这个函数检查是否取消，然后抛出异常
            println("Emitting $i")
        }
    }

    @Test
    fun `test cancel flow check`() = runBlocking<Unit> {
//        simpleFLow7().collect { value ->
//            println("value=$value")
//            if (value == 3) cancel()
//        }
        //下面演示无法取消的情况，依然会打印1到5
        (1..5).asFlow().collect { value ->
            println("value=$value")
            if (value == 3) cancel()
        }
        //要如何被取消，调用可取消的函数cancellable
        (1..5).asFlow().cancellable().collect { value ->
            println("value=$value")
            if (value == 3) cancel()
        }
    }

    /**背压***/

    fun simpleFLow8() = flow<Int> {
        for (i in 1..5) {
            delay(100)
            emit(i)//ensureActive 会被这个函数检查是否取消，然后抛出异常
            println("Emitting $i ${Thread.currentThread().name}")
        }
    }


    @Test
    fun `test flow back pressure`() = runBlocking<Unit> {
        //演示存在背压时。把管道加长，让更多的流可以进入通过，从而减少堆积，减少执行时间
        val time = measureTimeMillis {
            simpleFLow8()
//                .buffer(50)//这个是异步的
//                .conflate()//跳过中间的值，处理最新的值 1-5 执行了1 3 5
                .collect {//collectLatest最后一个的输出
                    delay(300)
                    println("Collected $it ${Thread.currentThread().name}")
                }

        }
        println("Collected in $time ms")
    }
//    collector.block()相当于block.invoke(collector)

    @Test
    fun `test flow transform`() = runBlocking<Unit> {
        simpleFLow8()
            .transform { request->
                emit("request = ${request}")
            }
            .collect {//collectLatest最后一个的输出
                delay(300)
                println("Collected $it ${Thread.currentThread().name}")
            }
    }
    @Test
    fun `test flow take`() = runBlocking<Unit> {
        simpleFLow8()
            .take(2)
            .collect {//collectLatest最后一个的输出
                delay(300)
                println("Collected $it ${Thread.currentThread().name}")
            }
    }
}